package com.lwc.backendsinglestructure.queue.consumer;

import cn.hutool.core.util.StrUtil;
import com.lwc.backendsinglestructure.queue.enums.MessageQueueEnum;
import com.lwc.backendsinglestructure.queue.model.BaseQueueDTO;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.*;

/**
 * 抽象的 Redis 消息队列服务
 */
@Slf4j
public abstract class AbstractRedisQueueConsumer<T extends BaseQueueDTO> {
//    Redis服务
//    @Autowired
//    private RedisService redisService;


    /**
     * 默认没开启
     */
    private volatile boolean isInit = false;

    private volatile boolean isRun = true;

    /**
     * 线程池
     */
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 任务重新入队次数 默认 0次
     * 如果有需要子类可重写,更改默认值
     * @return
     */
    protected int getBackToQueueTime(){
        return 0;
    }

    /**
     * 队列的名称 由实现类去实现
     *
     * @return
     */
    protected abstract MessageQueueEnum getMessageQueue();

    protected String getQueueName(){
        return getMessageQueue().getQueue();
    }

    /**
     * 获取最大执行任务数量
     * 如果有需要子类可重写,更改默认值
     * @return
     */
    protected int getPoolSize(){
        return 1;
    };

    /**
     * 初始化
     */
    private void init() {
        // 1. 默认没开启
        if(isInit) {
            return;
        }

//        log.info("开始初始化消息队列，参数如下: \n" +
//                        "queue name: {} \n" +
//                        "queue timeout: {}\n" +
//                        "queue retry times: {}\n" +
//                        "queue back queue times: {}\n" +
//                        "queue pool size: {}",
//                getQueueName(), getTimeout(), getRetryTimes(), getBackQueueTimes(), getPoolSize()
//        );

        initThreadPool(getPoolSize());
//        Long startSize = redisService.lsize(getQueueName());
//        log.info("启动消息队列{}成功, 目前还有 {} 个元素没被消费", getQueueName(), startSize);
        isInit = true;
    }




    /**
     * TODO 线程池创建方式可以查询一下有无优化
     * TODO 似乎没有必要加锁
     * 初始化线程池
     * @param poolSize
     */
    private void initThreadPool(int poolSize){
        if(threadPoolExecutor == null){
            synchronized (this) {
                if (threadPoolExecutor == null) {
                    threadPoolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolSize);
                }
            }
        }
    }

    /**
     * 核心调度方法
     */
    public void schedule() {
        this.init();
        // 通过信号量去控制循环的速率
        final Semaphore semaphore = new Semaphore(getPoolSize());
        try {
            while(isRun){
                // 获取到信号量再执行,没获取到就阻塞等待
                semaphore.acquire();
                // TODO 执行具体的任务,从Redis队列中取出数据
//                String  data = redisService.rightPop(QUEUE_NAME());
                String originalData = "12";
                // 如果队列里面取出数据为空,就释放信号量
                if(StrUtil.isEmpty(originalData)){
                    semaphore.release();
                    continue;
                }
                // 线程池去异步执行
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        T data = null;
                        // 1. 将data转换为对象
                        try {
                            // 数据转换
                            data = convert(originalData);
                            // 处理数据
                            receive(data);
                        } catch (Exception ex) {
                            try {
                                // 先让任务重试
                                if(!retry(data)){
                                    // 再判断任务是否有重新入队的资格
                                    if(!backToQueue(data)){
                                        // 重试失败 以及 没有入队资格就告警
                                        error(data,ex);
                                    }
                                }
                            } catch (Exception innerEx) {
                                log.error("消费队列异常捕获失败，原因: ", innerEx);
                            }
                        } finally {
                            // 任务执行完释放信号量
                            semaphore.release();
                        }
                    }
                });
            }
        } catch (InterruptedException outerEx) {
            log.error("启动消息队列失败, 原因 ", outerEx);
        }
    }


    /**
     * 监听到错误 进行告警 由实现类去实现
     *
     * @param data
     * @param ex
     */
    protected abstract void error(T data, Exception ex);

    /**
     * 重新入队方法
     * @return
     */
    private boolean backToQueue(T data){
        if (data == null){
            return false;
        }
        if(data.getBackToQueueTimes() < getBackToQueueTime()){
            log.info("数据重试失败，开始重新放入队列，已归队次数{}", data.getBackToQueueTimes());
            data.setBackToQueueTimes(data.getBackToQueueTimes() + 1);
            // 满足归队条件，重新归队
            submit(data);
            return true;
        }
        return false;
    }

    /**
     * 重新入队方法
     * @return
     */
    private void submit(T data){
        String queueName = getQueueName();
//        redisService.lpush(queueName, data);
    }

    /**
     * 重试机制
     * @return
     */
    private boolean retry(T data){

        if(data == null){
            return false;
        }

        if(data.getRetryTimes() < 0){
            return false;
        }

        // 重试次数
        for(int i = 1; i <= retryTimes(); i++){
            data.setRetryTimes(i);
            try {
                log.info("{} 开启第 {} 次重试", data.getClass(), i);
                receive(data);
                return true;
            } catch (Exception e) {
                log.error("队列任务执行失败, 重试第 {} 次", i);
            }

        }
        return false;
    }

    

    /**
     * 默认不重试，需要重试子类重写此方法
     * @return 重试次数
     */
    protected int retryTimes(){
        return 0;
    }

    /**
     * 接收到消息去处理（由实现类去实现）
     * @param data
     */
    protected abstract void receive(T data);

    /**
     * 数据转换由实现类去实现，因为不同实现类对接收到数据的转换都是不同的
     * @param data
     * @return
     */
    protected abstract T convert(String data);

}
